package com.sunday.common.mq.kafka.study.spring.e6_Receiving_Messages_3_Batch;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.ContainerCustomizer;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

/**
 * https://docs.spring.io/spring-kafka/docs/current/reference/html/#receiving-messages
 * {@link org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration}
 */
@Slf4j
@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    /**
     * https://docs.spring.io/spring-kafka/docs/current/reference/html/#batch-listeners
     */
    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Object, Object>> kafkaBatchListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ObjectProvider<ConsumerFactory<Object, Object>> kafkaConsumerFactory,
            ObjectProvider<ContainerCustomizer<Object, Object, ConcurrentMessageListenerContainer<Object, Object>>> kafkaContainerCustomizer,
            KafkaProperties properties
    ) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, kafkaConsumerFactory
                .getIfAvailable(() -> new DefaultKafkaConsumerFactory<>(properties.buildConsumerProperties())));
        kafkaContainerCustomizer.ifAvailable(factory::setContainerCustomizer);
        /**
         * https://docs.spring.io/spring-kafka/docs/current/reference/html/#batch-listeners
         * 若要配置侦听器容器工厂以创建批处理侦听器，可以设置该属性。 以下示例演示如何执行此操作：batchListener
         * factory.setBatchListener(true) 和 @KafkaListener(batch = "true") 同理，设置其一就行
         * 本例子中的效果
         * 2023-11-16 15:11:52.413  INFO [,] 1576 --- [   myId_3-0-C-1] c.s.c.m.k.s.s.e.MQApi [60] : [data] ======= 2
         * 2023-11-16 15:11:52.413  INFO [,] 1576 --- [   myId_3-0-C-1] c.s.c.m.k.s.s.e.MQApi [60] : [data] ======= 2
         * 2023-11-16 15:11:52.414  INFO [,] 1576 --- [   myId_3-0-C-1] c.s.c.m.k.s.s.e.MQApi [60] : [data] ======= 2
         * 2023-11-16 15:11:52.414  INFO [,] 1576 --- [   myId_3-0-C-1] c.s.c.m.k.s.s.e.MQApi [60] : [data] ======= 2
         * 2023-11-16 15:11:52.414  INFO [,] 1576 --- [   myId_3-0-C-1] c.s.c.m.k.s.s.e.MQApi [60] : [data] ======= 2
         */
        factory.setBatchListener(true);
        /**
         * https://docs.spring.io/spring-kafka/docs/current/reference/html/#using-ConcurrentMessageListenerContainer
         * 创建实例数量
         * 收听多个主题时，默认分区分布可能不是您期望的。
         * 例如，如果您有三个主题，每个主题有五个分区，并且您想要使用 ，则只看到 5 个活动使用者，每个使用者从每个主题分配一个分区，
         * 其他 10 个使用者处于空闲状态。 这是因为默认的 Kafka 是 （参见它的 Javadoc）。
         * 对于此方案，您可能需要考虑改用 ，它将分区分布在所有使用者之间。 然后，为每个使用者分配一个主题或分区。
         * 要更改 ，可以在提供给 的属性中设置使用者属性 （）。concurrency=15
         * 2023-11-16 15:18:45.253  INFO [,] 9960 --- [   myId_3-1-C-1] c.s.c.m.k.s.s.e.MQApi [60] : [data] ======= 2
         * 2023-11-16 15:18:45.253  INFO [,] 9960 --- [   myId_3-2-C-1] c.s.c.m.k.s.s.e.MQApi [60] : [data] ======= 2
         * 2023-11-16 15:18:45.254  INFO [,] 9960 --- [   myId_3-2-C-1] c.s.c.m.k.s.s.e.MQApi [60] : [data] ======= 2
         * 2023-11-16 15:18:45.255  INFO [,] 9960 --- [   myId_3-0-C-1] c.s.c.m.k.s.s.e.MQApi [60] : [data] ======= 2
         * 2023-11-16 15:18:45.255  INFO [,] 9960 --- [   myId_3-0-C-1] c.s.c.m.k.s.s.e.MQApi [60] : [data] ======= 2
         */
        factory.setConcurrency(3);
        return factory;
    }

    @Bean
    public NewTopic topic3() {
        return TopicBuilder.name("topic3")
                .partitions(3)
                .replicas(1)
                .build();
    }

}
